package com.gin.spark

import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ListBuffer

/**
 * map() 会一条记录为单位进行操作
 * mapPartitions() mapPartitionsWithIndex() 以分区为单位进行操作
 *
 */
object L05_Partitions {

  def main(args: Array[String]): Unit = {
    //加载配置 获取spark上下文对象
    val conf = new SparkConf().setMaster("local").setAppName("L01")
    val sc = new SparkContext(conf)
    //只打印错误日志
    sc.setLogLevel("ERROR")

    val data = sc.parallelize(1 to 5, 2)

    //读取mysql中的数据进行加工
    //直接使用map会频繁创建销毁连接
    data.map(x => {
        println("--------mysql-connect--------")
        println(s"--------mysql-select-$x--------")
        println("--------close-mysql-connect--------")
        x
      }
    ).foreach(println)

    //方式一: oom问题
    data.mapPartitionsWithIndex(
      (partitionIndex, partitionIter) => {
        //spark是一个pipeline, 迭代器嵌套模式, 不应该用一个集合去实际存储数据
        val lb = new ListBuffer[String]
        println(s"--$partitionIndex--------mysql-connect--------")
        while (partitionIter.hasNext) {
          val value: Int = partitionIter.next()
          println(s"--$partitionIndex--------mysql-select-$value--------")
          lb.+=(value + "-selected")
        }
        println("--------close-mysql-connect--------")
        //返回自定义集合的迭代器(lb会存在内存积压)
        lb.iterator
      }
    ).foreach(println)

    println()
    println("-------- mysql 1 --------")
    println()


    //方式二: 性能优化
    //数据是获取一条加工一条, 不应该在程序内部进行缓存(内存积压,可能会导致oom)
    //所以在处理时应使用spark的迭代器,不会实际存储数据
    data.mapPartitionsWithIndex(
      (partitionIndex, partitionIter) => {
        new Iterator[String] {

          //创建mysql连接
          println(s"--$partitionIndex--------mysql-connect--------")

          //返回是否有下一个元素,并在没有元素时释放mysql连接
          override def hasNext: Boolean = if (!partitionIter.hasNext) {
            println("--------close-mysql-connect--------");
            false
          } else true

          override def next(): String = {
            val value: Int = partitionIter.next()
            println(s"--$partitionIndex--------mysql-select-$value--------")
            value + "-selected"
          }
        }
      }
    ).foreach(println)


    println()
    println("-------- mysql 2 --------")
    println()


  }

}
